<template>
  <div class="kafka-realtime">
    <el-card class="box-card">
      <div slot="header" class="clearfix">
        <span>Kafka实时消息</span>
      </div>
      
      <!-- Kafka消息实时展示区域 -->
      <div class="kafka-realtime-container">
        <h3 class="section-title">Kafka实时消息</h3>
        <div class="kafka-controls">
          <el-select v-model="selectedKafkaForMessages" placeholder="选择Kafka配置来接收消息" style="width: 70%;">
            <el-option
              v-for="config in kafkaConfigs"
              :key="config.id"
              :label="`${config.name} (${config.url}/${config.topic})`"
              :value="config.id">
              <span style="float: left">{{ config.name }}</span>
              <span style="float: right; color: #8492a6; font-size: 13px">{{ config.topic }}</span>
            </el-option>
          </el-select>
          <el-button 
            type="primary" 
            :disabled="!selectedKafkaForMessages || isWebSocketConnected"
            @click="connectWebSocket">
            {{ isWebSocketConnected ? '已连接' : '连接并订阅' }}
          </el-button>
          <el-button 
            type="danger" 
            :disabled="!isWebSocketConnected"
            @click="disconnectWebSocket">
            断开连接
          </el-button>
        </div>
        
        <!-- 消息过滤控件 -->
        <div class="filter-controls">
          <el-input
            v-model="waybillNoFilter"
            placeholder="输入运单号进行过滤"
            clearable
            style="width: 300px;"
            prefix-icon="el-icon-search">
            <template slot="prepend">运单号</template>
          </el-input>
          <span class="filter-info">{{ filteredMessages.length }}/{{ kafkaMessages.length }} 条消息</span>
        </div>
        
        <!-- 修改消息展示区域为表格形式 -->
        <div class="message-display">
          <!-- 始终显示表格，包括表头 -->
          <div class="messages-header" v-if="kafkaMessages.length > 0">
            <span>共 {{ filteredMessages.length }} 条消息{{ waybillNoFilter ? '（已过滤）' : '' }}</span>
            <el-button type="text" @click="clearMessages">清空消息</el-button>
          </div>
          
          <!-- 表格展示区域 -->
          <div class="message-table-container">
            <el-table
              :data="filteredMessages"
              border
              style="width: 100%"
              height="350"
              @row-click="selectMessage">
              <!-- 时间和偏移量列 -->
              <el-table-column
                prop="timestamp"
                label="时间"
                width="100">
              </el-table-column>
              <el-table-column
                prop="offset"
                label="Offset"
                width="100">
              </el-table-column>
              
              <!-- 动态生成列 -->
              <el-table-column
                v-for="column in keyMapColumns"
                :key="column.key"
                :prop="'parsedData.' + column.key"
                :label="column.name"
                min-width="150"
                show-overflow-tooltip>
                <template slot-scope="scope">
                  {{ getColumnValue(scope.row, column.key) }}
                </template>
              </el-table-column>
            </el-table>
          </div>
          
          <!-- 空消息提示 -->
          <div v-if="kafkaMessages.length === 0" class="empty-message-overlay">
            暂无消息，连接Kafka配置后将在此显示实时消息
          </div>
          <div v-else-if="filteredMessages.length === 0 && waybillNoFilter" class="empty-message-overlay">
            没有找到符合过滤条件的消息
          </div>
        </div>
      </div>
    </el-card>
    
    <!-- 消息详情弹窗 -->
    <el-dialog :visible.sync="messageDetailVisible" title="消息详情" width="70%">
      <div v-if="selectedMessage">
        <div class="message-meta">
          <el-row :gutter="20">
            <el-col :span="12">
              <span class="meta-label">时间戳:</span>
              <span class="meta-value">{{ selectedMessage.timestamp }}</span>
            </el-col>
            <el-col :span="12">
              <span class="meta-label">Offset:</span>
              <span class="meta-value">{{ selectedMessage.offset }}</span>
            </el-col>
          </el-row>
        </div>
        
        <div class="message-content-detail">
          <h4>消息内容</h4>
          <div v-if="parsedMessageContent">
            <el-table :data="messageKeyValuePairs" border style="width: 100%" max-height="400">
              <el-table-column prop="key" label="Key" width="200" show-overflow-tooltip>
                <template slot-scope="scope">
                  <span class="key-text">{{ scope.row.key }}</span>
                </template>
              </el-table-column>
              <el-table-column prop="chineseName" label="中文字段" width="150" show-overflow-tooltip>
                <template slot-scope="scope">
                  <span class="chinese-name-text">{{ scope.row.chineseName || '-' }}</span>
                </template>
              </el-table-column>
              <el-table-column prop="value" label="Value" show-overflow-tooltip>
                <template slot-scope="scope">
                  <span class="value-text" :class="getValueClass(scope.row.type)">{{ scope.row.displayValue }}</span>
                </template>
              </el-table-column>
              <el-table-column prop="type" label="类型" width="100">
                <template slot-scope="scope">
                  <el-tag size="mini" :type="getTypeTagType(scope.row.type)">{{ scope.row.type }}</el-tag>
                </template>
              </el-table-column>
            </el-table>
          </div>
          <div v-else class="raw-content">
            <h5>原始内容（JSON解析失败）</h5>
            <pre class="raw-text">{{ selectedMessage.content }}</pre>
          </div>
        </div>
      </div>
      <div slot="footer">
        <el-button @click="messageDetailVisible = false">关闭</el-button>
      </div>
    </el-dialog>
  </div>
</template>

<script>
import { kafkaConfigApi } from '../api';

export default {
  name: 'KafkaRealtime',
  data() {
    return {
      kafkaConfigs: [],
      selectedKafkaForMessages: null,
      webSocket: null,
      isWebSocketConnected: false,
      kafkaMessages: [],
      selectedMessage: null,
      messageDetailVisible: false,
      waybillNoFilter: '',   // 运单号过滤
      parsedMessageContent: null,  // 解析后的消息内容
      messageKeyValuePairs: [],     // 消息的key-value对数组
      keyMapColumns: [] // 用于存储Kafka消息的列定义
    }
  },
  
  computed: {
    /**
     * 根据过滤条件筛选后的消息列表
     * 
     * @return {Array} 过滤后的消息数组
     */
    filteredMessages() {
       if (!this.waybillNoFilter) {
         return this.kafkaMessages;
       }
       
       return this.kafkaMessages.filter(message => 
         this.messageMatchesWaybillFilter(message)
       );
     }
  },
  created() {
    // 加载Kafka配置列表
    this.loadKafkaConfigs();
    
    // 加载key映射列表
    this.loadKeyMapData();
  },
  methods: {
    loadKafkaConfigs() {
      kafkaConfigApi.listKafkaConfigs()
        .then(response => {
          if (response.code === 0) {
            this.kafkaConfigs = response.data;
          } else {
            this.$message.error(`获取Kafka配置列表失败: ${response.message}`);
          }
        })
        .catch(error => {
          console.error('获取Kafka配置列表时出错:', error);
          this.$message.error('获取Kafka配置列表失败，请检查网络连接或服务器状态');
        });
    },
    
    // Kafka消息相关方法
    /**
     * 建立WebSocket连接
     * 
     * 功能说明：
     * 1. 验证是否选择了Kafka配置
     * 2. 构建WebSocket连接URL
     * 3. 创建WebSocket实例并设置事件处理器
     * 4. 处理连接状态和消息接收
     */
    connectWebSocket() {
      // 检查是否选择了Kafka配置
      if (!this.selectedKafkaForMessages) {
        this.$message.warning('请先选择Kafka配置');
        return;
      }
      
      // 根据选择的配置ID查找对应的配置对象
      const selectedConfig = this.kafkaConfigs.find(config => config.id === this.selectedKafkaForMessages);
      if (!selectedConfig) {
        this.$message.error('无法找到选中的Kafka配置');
        return;
      }
      
      // 构建WebSocket连接URL
      // 根据当前页面的协议自动选择ws或wss协议
      const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
      const host = window.location.hostname;
      // WebSocket服务运行在8001端口，路径包含配置ID用于后端识别
      const wsUrl = `${protocol}//${host}:8001/ws/kafka/${selectedConfig.id}/`;
      
      console.log('尝试连接到WebSocket:', wsUrl);
      
      // 创建WebSocket连接实例
      try {
        this.webSocket = new WebSocket(wsUrl);
      } catch (e) {
        console.error('创建WebSocket连接时出错:', e);
        this.$message.error('创建WebSocket连接时出错: ' + e.message);
        return;
      }
      
      // 设置WebSocket连接成功事件处理器
      this.webSocket.onopen = () => {
        console.log('WebSocket连接已建立');
        this.isWebSocketConnected = true;  // 更新连接状态
        this.$message.success('WebSocket连接已建立');
      };
      
      // 设置WebSocket消息接收事件处理器
      this.webSocket.onmessage = (event) => {
        try {
          console.log('收到WebSocket消息:', event.data);
          // 解析从后端接收到的JSON消息
          const message = JSON.parse(event.data);
          
          // 检查是否是错误消息
          if (message.error) {
            this.$message.error(`WebSocket错误: ${message.error}`);
            return;
          }
          
          // 解析消息内容
          let parsedData = {};
          try {
            parsedData = JSON.parse(message.value);
          } catch (e) {
            console.warn('消息不是有效的JSON格式:', e);
            // 如果不是JSON格式，尝试直接使用
            if (typeof message.value === 'object') {
              parsedData = message.value;
            }
          }
          
          // 构造前端显示的消息对象并添加到列表顶部
          this.kafkaMessages.unshift({
            timestamp: new Date().toLocaleTimeString(),  // 本地时间戳
            offset: message.offset,  // Kafka消息偏移量
            content: JSON.stringify(message.value, null, 2),  // 格式化的JSON内容
            parsedData: parsedData  // 解析后的JSON数据对象
          });
          
          // 限制消息列表长度为100条，避免内存占用过多
          if (this.kafkaMessages.length > 100) {
            this.kafkaMessages = this.kafkaMessages.slice(0, 100);
          }
        } catch (e) {
          console.error('解析消息失败:', e);
          this.$message.error('解析WebSocket消息失败: ' + e.message);
        }
      };
      
      // 设置WebSocket连接关闭事件处理器
      this.webSocket.onclose = (event) => {
        console.log('WebSocket连接已关闭', event);
        this.isWebSocketConnected = false;  // 更新连接状态
        this.$message.info('WebSocket连接已关闭');
      };
      
      // 设置WebSocket错误事件处理器
      this.webSocket.onerror = (error) => {
        console.error('WebSocket错误:', error);
        this.$message.error('WebSocket连接出错，请检查控制台获取详细信息');
        this.isWebSocketConnected = false;  // 更新连接状态
      };
    },
    
    /**
     * 断开WebSocket连接
     * 
     * 主动关闭WebSocket连接并清理相关状态
     */
    disconnectWebSocket() {
      if (this.webSocket) {
        this.webSocket.close();  // 关闭WebSocket连接
        this.webSocket = null;   // 清空WebSocket实例
        this.isWebSocketConnected = false;  // 更新连接状态
      }
    },
    
    /**
     * 清空消息列表
     * 
     * 清除所有已接收的Kafka消息
     */
    clearMessages() {
      this.kafkaMessages = [];
    },
    
    /**
     * 根据指定的key路径从JSON对象中获取值
     * 
     * @param {Object} obj - 要搜索的对象
     * @param {string} path - 属性路径，如 'user.name' 或 'data.id'
     * @return {*} 找到的值，如果路径不存在则返回undefined
     */
    getValueByPath(obj, path) {
      if (!path || !obj) return undefined;
      
      const keys = path.split('.');
      let current = obj;
      
      for (const key of keys) {
        if (current === null || current === undefined || typeof current !== 'object') {
          return undefined;
        }
        current = current[key];
      }
      
      return current;
    },
    
    /**
     * 检查消息是否匹配运单号过滤条件
     * 
     * @param {Object} message - 要检查的消息对象
     * @return {boolean} 是否匹配
     */
    messageMatchesWaybillFilter(message) {
      // 如果没有设置运单号过滤条件，则显示所有消息
      if (!this.waybillNoFilter) return true;
      
      try {
        // 使用已解析的数据对象
        const messageObj = message.parsedData || JSON.parse(message.content);
        
        // 获取运单号字段的值
        const waybillNo = this.getValueByPath(messageObj, 'waybillNo');
        
        if (waybillNo !== undefined) {
          // 将运单号转换为字符串并进行模糊匹配
          const waybillStr = String(waybillNo).toLowerCase();
          const filterStr = this.waybillNoFilter.toLowerCase();
          return waybillStr.includes(filterStr);
        }
        
        return false;
      } catch (e) {
        // 如果JSON解析失败，返回false
        console.warn('解析消息内容失败:', e);
        return false;
      }
    },
    
    /**
     * 选择消息查看详情
     * 
     * @param {Object} row - 要查看的消息行对象
     */
    selectMessage(row) {
      this.selectedMessage = row;
      this.parseMessageContent();
      this.messageDetailVisible = true;  // 显示消息详情弹窗
    },
    
    /**
     * 解析消息内容为key-value对
     */
    parseMessageContent() {
      if (!this.selectedMessage) {
        this.parsedMessageContent = null;
        this.messageKeyValuePairs = [];
        return;
      }
      
      try {
        // 使用已解析的数据对象或重新解析JSON
        const jsonObj = this.selectedMessage.parsedData || JSON.parse(this.selectedMessage.content);
        this.parsedMessageContent = jsonObj;
        this.messageKeyValuePairs = this.flattenObject(jsonObj);
      } catch (e) {
        this.parsedMessageContent = null;
        this.messageKeyValuePairs = [];
      }
    },
    
    /**
     * 将嵌套对象扁平化为key-value对数组
     * 
     * @param {Object} obj - 要扁平化的对象
     * @param {string} prefix - 键名前缀
     * @return {Array} key-value对数组
     */
    flattenObject(obj, prefix = '') {
      const result = [];
      
      for (const key in obj) {
        if (Object.prototype.hasOwnProperty.call(obj, key)) {
          const fullKey = prefix ? `${prefix}.${key}` : key;
          const value = obj[key];
          const type = this.getValueType(value);
          const chineseName = this.getChineseName(fullKey);
          
          if (value !== null && typeof value === 'object' && !Array.isArray(value)) {
            // 如果是对象，递归扁平化
            result.push({
              key: fullKey,
              value: value,
              displayValue: '[Object]',
              type: 'object',
              chineseName: chineseName
            });
            result.push(...this.flattenObject(value, fullKey));
          } else {
            result.push({
              key: fullKey,
              value: value,
              displayValue: this.getMappedValue(fullKey, value),
              type: type,
              chineseName: chineseName
            });
          }
        }
      }
      
      return result;
    },
    
    /**
     * 获取值的类型
     * 
     * @param {*} value - 要检查的值
     * @return {string} 值的类型
     */
    getValueType(value) {
      if (value === null) return 'null';
      if (Array.isArray(value)) return 'array';
      if (typeof value === 'object') return 'object';
      if (typeof value === 'string') return 'string';
      if (typeof value === 'number') return 'number';
      if (typeof value === 'boolean') return 'boolean';
      return 'unknown';
    },
    
    /**
     * 格式化显示值
     * 
     * @param {*} value - 要格式化的值
     * @return {string} 格式化后的字符串
     */
    formatValue(value) {
      if (value === null) return 'null';
      if (value === undefined) return '';  // undefined显示为空
      if (Array.isArray(value)) {
        return `[${value.length} items] ${JSON.stringify(value)}`;
      }
      if (typeof value === 'object') {
        return JSON.stringify(value, null, 2);
      }
      if (typeof value === 'string') {
        return value;  // 字符串不加引号
      }
      return String(value);
    },
    
    /**
     * 获取值类型对应的CSS类
     * 
     * @param {string} type - 值类型
     * @return {string} CSS类名
     */
    getValueClass(type) {
      return `value-${type}`;
    },
    
    /**
     * 获取类型标签的样式
     * 
     * @param {string} type - 值类型
     * @return {string} Element UI标签类型
     */
    getTypeTagType(type) {
      const typeMap = {
        'string': '',
        'number': 'success',
        'boolean': 'warning',
        'object': 'info',
        'array': 'info',
        'null': 'danger'
      };
      return typeMap[type] || '';
    },

    /**
     * 获取表格列的值
     * 
     * @param {Object} row - 当前行数据
     * @param {string} key - 列的key
     * @return {string} 列的值
     */
    getColumnValue(row, key) {
      const value = this.getValueByPath(row.parsedData, key);
      return this.getMappedValue(key, value);
    },

    /**
     * 根据英文key获取对应的中文名称
     * 
     * @param {string} key - 英文key
     * @return {string} 中文名称，如果找不到则返回空字符串
     */
    getChineseName(key) {
      const keyMap = this.keyMapColumns.find(item => item.key === key);
      return keyMap ? keyMap.name : '';
    },

    /**
     * 根据英文key和对应的值获取码值映射后的显示文本
     * 
     * @param {string} key - 英文key
     * @param {*} value - 对应的值
     * @return {string} 映射后的显示文本
     */
    getMappedValue(key, value) {
      const keyMap = this.keyMapColumns.find(item => item.key === key);
      
      // 如果找不到对应的keyMap或者value字段为空，返回原值
      if (!keyMap || !keyMap.value) {
        return this.formatValue(value);
      }
      
      try {
        // 解析value字段的JSON
        const valueMapping = JSON.parse(keyMap.value);
        
        // 将值转换为字符串进行匹配
        const valueStr = String(value);
        
        // 如果找到对应的键，返回对应的值
         if (Object.prototype.hasOwnProperty.call(valueMapping, valueStr)) {
           return valueMapping[valueStr];
         }
         
         // 如果找不到对应的键，返回other键对应的值
         if (Object.prototype.hasOwnProperty.call(valueMapping, 'other')) {
           return valueMapping.other;
         }
        
        // 如果连other键都没有，返回原值
        return this.formatValue(value);
      } catch (e) {
        // JSON解析失败，返回原值
        console.warn('解析value字段JSON失败:', e);
        return this.formatValue(value);
      }
    },

    // 添加加载 k_key_map 数据的方法
    loadKeyMapData() {
      kafkaConfigApi.getKeyMapList()
        .then(response => {
          if (response.code === 0) {
            this.keyMapColumns = response.data.map(item => ({
              name: item.name,
              key: item.key,
              value: item.value
            }));
            console.log('获取到key映射列表:', this.keyMapColumns);
          } else {
            this.$message.error(`获取key映射列表失败: ${response.message}`);
          }
        })
        .catch(error => {
          console.error('获取key映射列表时出错:', error);
          this.$message.error('获取key映射列表失败，请检查网络连接或服务器状态');
        });
    }
  },
  
  /**
   * Vue组件销毁前的生命周期钩子
   * 
   * 在组件即将被销毁时执行清理工作，防止内存泄漏。
   * 主要清理：
   * - 关闭WebSocket连接
   * - 释放相关资源
   */
  beforeDestroy() {
    // 组件销毁前关闭WebSocket连接，避免连接泄漏
    if (this.webSocket) {
      this.webSocket.close();
    }
  }
}
</script>

<style scoped>
.kafka-realtime {
  width: 100%;
  margin: 0;
  padding: 20px;
  box-sizing: border-box;
}

/* Kafka消息展示区域样式 */
.kafka-realtime-container {
  margin-top: 30px;
  border-top: 1px solid #ebeef5;
  padding-top: 20px;
}

.section-title {
  margin-bottom: 15px;
  color: #303133;
}

.kafka-controls {
  display: flex;
  align-items: center;
  gap: 10px;
  margin-bottom: 15px;
}

.filter-controls {
  display: flex;
  align-items: center;
  gap: 15px;
  margin-bottom: 15px;
  padding: 10px;
  background-color: #f8f9fa;
  border-radius: 4px;
}

.filter-info {
  font-size: 12px;
  color: #909399;
  white-space: nowrap;
}

.message-display {
  position: relative;
  border: 1px solid #ebeef5;
  border-radius: 4px;
  min-height: 200px;
  max-height: 450px;
  overflow: hidden;
}

.messages-header {
  display: flex;
  justify-content: space-between;
  align-items: center;
  padding: 10px 15px;
  border-bottom: 1px solid #ebeef5;
  background-color: #f5f7fa;
}

/* 表格容器样式，支持横向滚动 */
.message-table-container {
  width: 100%;
  overflow: hidden;
}

/* 确保表格元素样式正确 */
.message-table-container >>> .el-table {
  width: 100% !important;
  height: 100%;
}

/* 确保表格可以横向滚动 */
.message-table-container >>> .el-table__body-wrapper {
  overflow-x: auto !important;
}

/* 确保表格内容正常显示 */
.message-table-container >>> .el-table__header,
.message-table-container >>> .el-table__body {
  width: 100% !important;
  table-layout: fixed !important;
}

.message-table-container >>> .el-table .cell {
  white-space: nowrap;
  overflow: hidden;
  text-overflow: ellipsis;
}

.empty-message {
  color: #909399;
  text-align: center;
  padding: 20px;
}

.empty-message-overlay {
  position: absolute;
  top: 50%;
  left: 50%;
  transform: translate(-50%, -50%);
  color: #909399;
  text-align: center;
  padding: 20px;
  background-color: rgba(255, 255, 255, 0.9);
  border-radius: 4px;
  z-index: 10;
}

/* 消息详情弹窗样式 */
.message-meta {
  margin-bottom: 20px;
  padding: 15px;
  background-color: #f8f9fa;
  border-radius: 4px;
}

.meta-label {
  font-weight: bold;
  color: #606266;
  margin-right: 8px;
}

.meta-value {
  color: #303133;
}

.message-content-detail h4 {
  margin-bottom: 15px;
  color: #303133;
  border-bottom: 1px solid #ebeef5;
  padding-bottom: 8px;
}

.key-text {
  font-weight: bold;
  color: #409eff;
}

.chinese-name-text {
  color: #67c23a;
  font-weight: 500;
}

.value-text {
  word-break: break-all;
}

.value-string {
  color: #67c23a;
}

.value-number {
  color: #e6a23c;
}

.value-boolean {
  color: #f56c6c;
  font-weight: bold;
}

.value-object {
  color: #909399;
  font-style: italic;
}

.value-array {
  color: #909399;
}

.value-null {
  color: #f56c6c;
  font-style: italic;
}

.raw-content {
  margin-top: 15px;
}

.raw-content h5 {
  color: #f56c6c;
  margin-bottom: 10px;
}

.raw-text {
  background-color: #f5f5f5;
  padding: 15px;
  border-radius: 4px;
  max-height: 300px;
  overflow-y: auto;
  font-size: 12px;
  color: #606266;
}
</style>